package com.niit.offline.rec

import org.apache.spark.SparkConf
import org.apache.spark.mllib.recommendation.{ALS, Rating}
import org.apache.spark.sql.SparkSession
import org.jblas.DoubleMatrix

case class ProductRating( userId: Int, productId: Int, score: Double, timestamp:Int )// 定义用户-商品-评分的样例类
case class MongoConfig( uri: String, db: String ) // 定义MongoConfig样例类
case class Recommendation( productId: Int, score: Double ) // 定义标准推荐对象
case class UserRecs( userId: Int, recs: Seq[Recommendation] ) // 定义用户的推荐列表
case class ProductRecs( productId: Int, recs: Seq[Recommendation] ) // 定义商品相似度列表

object OfflineRecommender {

  val MONGODB_RATING_COLLECTION = "Rating" //定义要读取MongoDB中的表
  val MONGODB_USER_RECS_COLLECTION = "UserRecs" // 定义根据ALS算法得到的用户推荐列表数据在Mongo中存放的表
  val USER_MAX_RECOMMENDATION_SIZE = 20 // 定义最大只取出用户推荐列表中的前20个相似度的用户
  val MONGODB_PRODUCT_RECS_COLLECTION = "ProductRecs" // 定义根据ALS算法得到的商品相似度矩阵数据在Mongo中存放的表

  def main(args: Array[String]): Unit = {
    // 1.定义MongoDB连接属性
    val config = Map(
      "spark.cores" -> "local[*]",
      "mongo.uri" -> "mongodb://fooadmin:123456@hadoop001:27017/bigdata",
      "mongo.db" -> "bigdata"
    )
    // 2.创建SparkConf对象
    val sparkConf = new SparkConf().setAppName("OfflineRecommender").setMaster(config("spark.cores"))
    // 3.创建SparkSession对象
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    // 4.引入spark隐式调用相关的类
    import spark.implicits._
    // 5.创建隐式的MongoConfig对象
    implicit val mongoConfig = MongoConfig(config("mongo.uri"),config("mongo.db"))
    // 6.加载数据
    val ratingRDD = spark
      .read
      .option("uri", mongoConfig.uri)
      .option("collection", MONGODB_RATING_COLLECTION)
      .format("com.mongodb.spark.sql")
      .load()
      .as[ProductRating]
      .rdd
      .map(
        rating => (rating.userId, rating.productId, rating.score)
      ).cache()
    // 7.对数据进行预处理
    val userRDD = ratingRDD.map(_._1).distinct() // 7.1 提取出ratingRDD中的所有用户数据
    val productRDD = ratingRDD.map(_._2).distinct()  // 7.2 提取出ratingRDD中的所有商品数据
    // 8.核心业务逻辑
    // 8.1 实现用户推荐列表
    // 思路：1.定义训练模型；2.获取预测评分矩阵，得到用户推荐列表；3. 从用户推荐列表中提取前20个
    // 1.定义训练模型；
    val trainData = ratingRDD.map(x => Rating(x._1, x._2, x._3))
    // 拍脑袋给出的值,即模型所需参数：rank 代表特征维度，iterations代表交叉迭代次数，lambda代表正则化参数
    val (rank,iterations,lambda) = (5,10,0.1)
    val model = ALS.train(trainData, rank, iterations, lambda)
    // 2.获取预测评分矩阵，得到用户推荐列表；
    // 2.1 用userRDD和productRDD做一个笛卡尔积，得到空的userProductsRDD表示的评分矩阵
    val userProductsRDD = userRDD.cartesian(productRDD)
    val preRatings = model.predict(userProductsRDD)
    // 3. 从用户推荐列表中提取前20个,数据格式(userId,Seq[Recommendation]
    val userRecsDF = preRatings.filter(_.rating > 0).map(
      rating => (rating.user, (rating.product, rating.rating))
    )
      .groupByKey()
      .map {
        case (userId, recs) =>
          UserRecs(userId,
            recs.toList
              .sortWith(_._2 > _._2)                  // 按照rating评分进行降序排
              .take(USER_MAX_RECOMMENDATION_SIZE)     // 取出前20个符合条件的用户
              .map(x => Recommendation(x._1, x._2)))  // 将map中的元素变成样例类Recommendation
      }.toDF()
    // 4. 写入mongodb
    userRecsDF.write
        .option("uri",mongoConfig.uri)
        .option("collection",MONGODB_USER_RECS_COLLECTION)
        .mode("overwrite")
        .format("com.mongodb.spark.sql")
        .save()
    // 8.2 计算商品相似度矩阵
    //获取商品的特征矩阵，数据格式 RDD[(scala.Int, scala.Array[scala.Double])]
    val productFeatures = model.productFeatures.map{case (productId,features) =>
      (productId, new DoubleMatrix(features))
    }
    // 计算笛卡尔积并过滤合并
    val productRecs = productFeatures.cartesian(productFeatures)
      .filter{case (a,b) => a._1 != b._1}
      .map{case (a,b) =>
        val simScore = this.consinSim(a._2,b._2) // 求余弦相似度
        (a._1,(b._1,simScore))
      }.filter(_._2._2 > 0.6)
      .groupByKey()
      .map{case (productId,items) =>
        ProductRecs(productId,items.toList.map(x => Recommendation(x._1,x._2)))
      }.toDF()

    productRecs
      .write
      .option("uri", mongoConfig.uri)
      .option("collection",MONGODB_PRODUCT_RECS_COLLECTION)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()
    // 9.停止spark
    spark.stop()
  }

  //计算两个商品之间的余弦相似度
  def consinSim(product1: DoubleMatrix, product2:DoubleMatrix) : Double ={
    product1.dot(product2) / ( product1.norm2()  * product2.norm2() )
  }

}
